package com.zhu.util;

import com.zhu.config.ClusterParametersConfig;
import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.Serializable;
import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

/**
 * Flink没有提供 直接写入ClickHouse 的方法 需要自定义Sink
 */
public class ClickHouseUtil implements Serializable {

    public static <T> SinkFunction<T> getClickHouseSinkFunction(String sql) throws IOException {

        Properties properties = PropertiesUtil.getProperties("D:\\java\\NLP\\Flink-realtimeProcess\\src\\main\\resources\\jdbc.properties");

        return JdbcSink.<T>sink(sql, new JdbcStatementBuilder<T>() {
            @Override
            @SneakyThrows
            public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                Class<?> tClz = t.getClass();
                Field[] declaredFields = tClz.getDeclaredFields(); //获取所有属性
                //判断如果属性带有TransSink 注解就不会被序列化
                int offset = 0;
                for (int i = 0; i < declaredFields.length; i++) {
                    Field declaredField = declaredFields[i];
                    declaredField.setAccessible(true);
                    MyTransientSink annotation = declaredField.getAnnotation(MyTransientSink.class);
                    if(annotation != null){
                        offset ++;
                        continue;
                    }
                    Object value = declaredField.get(t);
                    preparedStatement.setObject(i + 1 - offset,value);
                }
            }
        }, JdbcExecutionOptions.builder()
                .withBatchSize(5)
                .withBatchIntervalMs(1000L)
                .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(properties.getProperty(ClusterParametersConfig.CLICKHOUSE_CONNECT_DRIVER))
                        .withUrl(properties.getProperty(ClusterParametersConfig.CLICKHOUSE_CONNECT_URL))
                        .withUsername(properties.getProperty(ClusterParametersConfig.CLICKHOUSE_CONNECT_USERNAME))
                        .withPassword(properties.getProperty(ClusterParametersConfig.CLICKHOUSE_CONNECT_PASSWORD))
                        .build());
    }
}
